package org.myfram.flink.flinkonjar.domain.process.kafkaJoinHbase;

import lombok.extern.slf4j.Slf4j;
import org.apache.flink.streaming.api.functions.co.KeyedBroadcastProcessFunction;
import org.apache.flink.util.Collector;
import org.myfram.flink.flinkonjar.application.job.KafkaJoinHbaseJob;
import org.myfram.flink.flinkonjar.domain.entity.KafkaJoinHBase.HBaseDimensionTable;
import org.myfram.flink.flinkonjar.domain.entity.KafkaJoinHBase.KafkaHBaseJoinedResult;
import org.myfram.flink.flinkonjar.domain.entity.KafkaJoinHBase.KafkaSourceUserEvent;

@Slf4j
public class BroadcastJoinFunction extends KeyedBroadcastProcessFunction<String, KafkaSourceUserEvent, HBaseDimensionTable, KafkaHBaseJoinedResult> {

    @Override
    public void processElement(KafkaSourceUserEvent kafkaSourceUserEvent, KeyedBroadcastProcessFunction<String, KafkaSourceUserEvent, HBaseDimensionTable, KafkaHBaseJoinedResult>.ReadOnlyContext readOnlyContext, Collector<KafkaHBaseJoinedResult> collector) throws Exception {
        HBaseDimensionTable dimensionData = readOnlyContext.getBroadcastState(KafkaJoinHbaseJob.DIMENSION_STATE_DESCRIPTOR).get(kafkaSourceUserEvent.getId());
        if (dimensionData != null) {
            KafkaHBaseJoinedResult result = new KafkaHBaseJoinedResult();
            result.setId(kafkaSourceUserEvent.getId());
            result.setName(dimensionData.getName());
            collector.collect(result);
            log.info("维表关联成功 - Kafka ID: {}, 维表Name: {}", kafkaSourceUserEvent.getId(), dimensionData.getName());
        } else {
            log.warn("未找到匹配的维表数据 - Kafka ID: {}", kafkaSourceUserEvent.getId());
        }
    }

    @Override
    public void processBroadcastElement(HBaseDimensionTable hBaseDimensionTable, KeyedBroadcastProcessFunction<String, KafkaSourceUserEvent, HBaseDimensionTable, KafkaHBaseJoinedResult>.Context context, Collector<KafkaHBaseJoinedResult> collector) throws Exception {
        context.getBroadcastState(KafkaJoinHbaseJob.DIMENSION_STATE_DESCRIPTOR).put(hBaseDimensionTable.getId(), hBaseDimensionTable);
        log.info("更新维表缓存 - ID: {}, Name: {}", hBaseDimensionTable.getId(), hBaseDimensionTable.getName());
    }

}